package com.lefeng.cockpit.datax.thread;

import com.lefeng.cockpit.common.bean.po.job.JobInfoPO;
import com.lefeng.cockpit.datax.conf.JobAdminConfig;
import com.lefeng.cockpit.datax.cron.CronExpression;
import com.lefeng.cockpit.datax.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Job scheduling helper responsible for managing job schedules and triggers.
 */
public class JobScheduleHelper {

    private static final Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
    private static final long PRE_READ_MS = 5000;  // Pre-read period in milliseconds

    private static final JobScheduleHelper INSTANCE = new JobScheduleHelper();
    private Thread scheduleThread;
    private Thread ringThread;
    private volatile boolean scheduleThreadToStop = false;
    private volatile boolean ringThreadToStop = false;
    private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

    private JobScheduleHelper() {
        // Private constructor to enforce singleton pattern
    }

    public static JobScheduleHelper getInstance() {
        return INSTANCE;
    }

    /**
     * Starts the scheduling and ring threads.
     */
    public void start() {
        startScheduleThread();
        startRingThread();
    }

    private void startScheduleThread() {
        scheduleThread = new Thread(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
            } catch (InterruptedException e) {
                handleThreadInterrupt(e, "scheduleThread");
            }
            logger.info("JobScheduleHelper scheduling thread started.");

            int preReadCount = calculatePreReadCount();

            while (!scheduleThreadToStop) {
                processSchedule(preReadCount);
            }
            logger.info("JobScheduleHelper scheduling thread stopped.");
        });
        configureAndStartThread(scheduleThread, "JobScheduleHelper Scheduling Thread");
    }

    private int calculatePreReadCount() {
        // Calculate pre-read count based on trigger pool sizes
        return (JobAdminConfig.getAdminConfig().getTriggerPoolFastMax() +
                JobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
    }

    private void processSchedule(int preReadCount) {
        long start = System.currentTimeMillis();
        boolean preReadSuccess = true;

        try (Connection conn = JobAdminConfig.getAdminConfig().getDataSource().getConnection();
             PreparedStatement preparedStatement = conn.prepareStatement("select * from dc_datax_job_lock where lock_name = 'schedule_lock' for update")) {

            conn.setAutoCommit(false);
            preparedStatement.execute();
            List<JobInfoPO> scheduleList = fetchScheduleList(preReadCount);

            if (scheduleList != null && !scheduleList.isEmpty()) {
                processJobs(scheduleList);
            } else {
                preReadSuccess = false;
            }

            conn.commit();

        } catch (Exception e) {
            if (!scheduleThreadToStop) {
                logger.error("Error in scheduling thread:", e);
            }
        }

        alignAndWait(start, preReadSuccess);
    }

    private List<JobInfoPO> fetchScheduleList(int preReadCount) throws SQLException {
        long nowTime = System.currentTimeMillis();
        return JobAdminConfig.getAdminConfig().getJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
    }

    private void processJobs(List<JobInfoPO> scheduleList) throws ParseException {
        long nowTime = System.currentTimeMillis();

        for (JobInfoPO jobInfo : scheduleList) {
            if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
                handleJobMissedTrigger(jobInfo);
            } else if (nowTime > jobInfo.getTriggerNextTime()) {
                handleJobTriggered(jobInfo);
            } else {
                handleJobPreRead(jobInfo);
            }
        }

        updateJobTriggers(scheduleList);
    }

    private void handleJobMissedTrigger(JobInfoPO jobInfo) throws ParseException {
        logger.warn("Job missed trigger, jobId = {}", jobInfo.getId());
        refreshNextValidTime(jobInfo, new Date());
    }

    private void handleJobTriggered(JobInfoPO jobInfo) throws ParseException {
        JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
        logger.debug("Job triggered, jobId = {}", jobInfo.getId());
        refreshNextValidTime(jobInfo, new Date());
        pushJobToRingIfNeeded(jobInfo);
    }

    private void handleJobPreRead(JobInfoPO jobInfo) throws ParseException {
        pushJobToRing(jobInfo);
        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    }

    private void pushJobToRingIfNeeded(JobInfoPO jobInfo) {
        if (jobInfo.getTriggerStatus() == 1 && System.currentTimeMillis() + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
            pushJobToRing(jobInfo);
        }
    }

    private void pushJobToRing(JobInfoPO jobInfo) {
        int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
        pushTimeRing(ringSecond, jobInfo.getId());
        logger.debug("Scheduled job {} pushed to time-ring at second {}", jobInfo.getId(), ringSecond);
    }

    private void updateJobTriggers(List<JobInfoPO> scheduleList) {
        for (JobInfoPO jobInfo : scheduleList) {
            JobAdminConfig.getAdminConfig().getJobInfoDao().scheduleUpdate(jobInfo);
        }
    }

    private void alignAndWait(long start, boolean preReadSuccess) {
        long cost = System.currentTimeMillis() - start;
        if (cost < 1000) {
            try {
                TimeUnit.MILLISECONDS.sleep((preReadSuccess ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
            } catch (InterruptedException e) {
                handleThreadInterrupt(e, "scheduleThread");
            }
        }
    }

    private void startRingThread() {
        ringThread = new Thread(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
            } catch (InterruptedException e) {
                handleThreadInterrupt(e, "ringThread");
            }
            logger.info("JobScheduleHelper ring thread started.");

            while (!ringThreadToStop) {
                processRing();
            }
            logger.info("JobScheduleHelper ring thread stopped.");
        });
        configureAndStartThread(ringThread, "JobScheduleHelper Ring Thread");
    }

    private void processRing() {
        List<Integer> ringItemData = new ArrayList<>();
        int nowSecond = Calendar.getInstance().get(Calendar.SECOND);

        for (int i = 0; i < 2; i++) {
            List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
            if (tmpData != null) {
                ringItemData.addAll(tmpData);
            }
        }

        if (!ringItemData.isEmpty()) {
            logger.debug("Time-ring beat at second {} with jobs: {}", nowSecond, ringItemData);
            ringItemData.forEach(jobId -> JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null));
        }

        try {
            TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
        } catch (InterruptedException e) {
            handleThreadInterrupt(e, "ringThread");
        }
    }

    private void refreshNextValidTime(JobInfoPO jobInfo, Date fromTime) throws ParseException {
        Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromTime);
        if (nextValidTime != null) {
            jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
            jobInfo.setTriggerNextTime(nextValidTime.getTime());
        } else {
            jobInfo.setTriggerStatus(0);
            jobInfo.setTriggerLastTime(0L);
            jobInfo.setTriggerNextTime(0L);
        }
    }

    private void pushTimeRing(int ringSecond, int jobId) {
        ringData.computeIfAbsent(ringSecond, k -> new ArrayList<>()).add(jobId);
        logger.debug("Added job {} to time-ring at second {}", jobId, ringSecond);
    }

    /**
     * Stops both the scheduling and ring threads.
     */
    public void toStop() {
        stopScheduleThread();
        stopRingThread();
        logger.info("JobScheduleHelper stopped.");
    }

    private void stopScheduleThread() {
        scheduleThreadToStop = true;
        waitForThreadTermination(scheduleThread);
        handlePendingRingData();
    }

    private void stopRingThread() {
        ringThreadToStop = true;
        waitForThreadTermination(ringThread);
    }

    private void waitForThreadTermination(Thread thread) {
        try {
            TimeUnit.SECONDS.sleep(1);  // Allow time for thread to finish current work
        } catch (InterruptedException e) {
            logger.error("Interrupted while waiting for thread to stop.", e);
        }
        if (thread.isAlive()) {
            thread.interrupt();
            try {
                thread.join();
            } catch (InterruptedException e) {
                logger.error("Error while joining thread.", e);
            }
        }
    }

    private void handlePendingRingData() {
        if (!ringData.isEmpty()) {
            try {
                TimeUnit.SECONDS.sleep(8);  // Allow time for pending jobs in the ring
            } catch (InterruptedException e) {
                logger.error("Interrupted while handling pending ring data.", e);
            }
        }
    }

    private void configureAndStartThread(Thread thread, String name) {
        thread.setDaemon(true);
        thread.setName(name);
        thread.start();
    }

    private void handleThreadInterrupt(InterruptedException e, String threadName) {
        if (!scheduleThreadToStop && !ringThreadToStop) {
            logger.error("Thread {} interrupted: {}", threadName, e.getMessage(), e);
        }
    }
}
